/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.connector.jdbc.statement;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.Date;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

/** Simple implementation of {@link FieldNamedPreparedStatement}. */
public class FieldNamedPreparedStatementImpl implements FieldNamedPreparedStatement {

    // statement是一个 PreparedStatement对象，用于执行预编译的 SQL语句。
    // 在 FieldNamedPreparedStatement 中，通过使用 PreparedStatement的 setObject()方法，
    // 并结合 indexMapping数组中的索引值，可以将参数值设置到 SQL语句中。
    private final PreparedStatement statement;
    
    // indexMapping是一个二维整型数组，用于记录 SQL语句中参数占位符的位置。
    // 在 FieldNamedPreparedStatement中，可以使用字符串形式的参数名来设置和获取参数值，
    // 而无需依赖于占位符的位置。在执行 SQL语句之前，
    // 需要将字符串形式的参数名转换为对应的占位符位置，并将这些位置存储在 indexMapping数组中。
    private final int[][] indexMapping;

    /**
     * 这是一个私有的构造函数，用于初始化 FieldNamedPreparedStatementImpl类的实例。
     * @param statement：表示一个预编译的 SQL语句对象，它表示要发送到数据库的 SQL语句。
     *                   这个对象可以有一个或多个 IN参数，这些参数在 SQL语句创建时就已经被定义好，
     *                   然后在执行时被填充。
     * @param indexMapping：表示一个二维整数数组，可能用于映射或者存储某些特定的信息，
     *                      具体的用途需要根据实际的上下文来确定。
     */
    private FieldNamedPreparedStatementImpl(PreparedStatement statement, int[][] indexMapping) {
        this.statement = statement;
        this.indexMapping = indexMapping;
    }

    @Override
    public void clearParameters() throws SQLException {
        statement.clearParameters();
    }

    /**
     * 这是一个方法 executeQuery，它用于执行查询操作并返回一个 ResultSet对象。
     * 
     * @return
     * @throws SQLException
     */
    @Override
    public ResultSet executeQuery() throws SQLException {
        // 在方法内部，调用了 statement.executeQuery()方法来执行预编译的 SQL语句，并返回查询结果集。
        // 如果在执行过程中发生了 SQLException异常，则需要在调用该方法时进行异常处理。
        return statement.executeQuery();
    }

    /**
     * 这是一个 addBatch方法，用于将当前参数设置添加到批处理中。
     * 通过批处理操作，可以一次性执行多个 SQL语句，从而提高数据库操作的效率。
     * 
     * @throws SQLException  
     *         如果在添加过程中发生了 SQLException 异常，则需要在调用该方法时进行异常处理。
     */
    @Override
    public void addBatch() throws SQLException {
        // 在方法内部，调用了statement.addBatch()方法
        // 将当前的参数设置添加到预编译的SQL语句的批处理列表中。
        statement.addBatch();
    }

    /**
     * 这是一个 executeBatch方法，用于执行批处理中的所有SQL语句，并返回一个整型数组。
     *
     * @return
     * @throws SQLException
     *         如果在执行过程中发生了 SQLException异常，则需要在调用该方法时进行异常处理。
     */
    @Override
    public int[] executeBatch() throws SQLException {
        // 在方法内部，调用了 statement.executeBatch()方法来执行预编译的 SQL语句的批处理列表，
        // 并返回一个整型数组，其中包含每个SQL语句执行后所影响的行数。
        return statement.executeBatch();
    }

    @Override
    public void setNull(int fieldIndex, int sqlType) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setNull(index, sqlType);
        }
    }

    @Override
    public void setBoolean(int fieldIndex, boolean x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setBoolean(index, x);
        }
    }

    @Override
    public void setByte(int fieldIndex, byte x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setByte(index, x);
        }
    }

    @Override
    public void setShort(int fieldIndex, short x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setShort(index, x);
        }
    }

    @Override
    public void setInt(int fieldIndex, int x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setInt(index, x);
        }
    }

    @Override
    public void setLong(int fieldIndex, long x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setLong(index, x);
        }
    }

    @Override
    public void setFloat(int fieldIndex, float x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setFloat(index, x);
        }
    }

    @Override
    public void setDouble(int fieldIndex, double x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setDouble(index, x);
        }
    }

    @Override
    public void setBigDecimal(int fieldIndex, BigDecimal x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setBigDecimal(index, x);
        }
    }

    @Override
    public void setString(int fieldIndex, String x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setString(index, x);
        }
    }

    @Override
    public void setBytes(int fieldIndex, byte[] x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setBytes(index, x);
        }
    }

    @Override
    public void setDate(int fieldIndex, Date x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setDate(index, x);
        }
    }

    @Override
    public void setTime(int fieldIndex, Time x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setTime(index, x);
        }
    }

    @Override
    public void setTimestamp(int fieldIndex, Timestamp x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setTimestamp(index, x);
        }
    }

    @Override
    public void setObject(int fieldIndex, Object x) throws SQLException {
        for (int index : indexMapping[fieldIndex]) {
            statement.setObject(index, x);
        }
    }

    @Override
    public void close() throws SQLException {
        statement.close();
    }

    // ----------------------------------------------------------------------------------------

    /**
     * 这是一个静态方法 prepareStatement，
     * 用于在给定的数据库连接上创建一个支持以 “字段名” 作为参数占位符的预编译语句对象。
     * 
     * @param connection：Connection对象，表示数据库连接；
     * @param sql：SQL语句字符串，表示 SQL语句；
     * @param fieldNames：字符串数组，表示查询结果集中需要返回的列名。
     * @return
     * @throws SQLException
     */
    public static FieldNamedPreparedStatement prepareStatement(
            Connection connection, String sql, String[] fieldNames) throws SQLException {
        
        // 在方法内部，首先通过调用 checkNotNull方法检查传入的参数是否为 null。
        checkNotNull(connection, "connection must not be null.");
        checkNotNull(sql, "sql must not be null.");
        checkNotNull(fieldNames, "fieldNames must not be null.");

        // 如果 SQL语句中包含?字符，则抛出异常。
        if (sql.contains("?")) {
            throw new IllegalArgumentException("SQL statement must not contain ? character.");
        }

        HashMap<String, List<Integer>> parameterMap = new HashMap<>();
        
        // 然后，通过调用 parseNamedStatement方法解析SQL语句，
        String parsedSQL = parseNamedStatement(sql, parameterMap);
        
        
        //
        // 目前，语句必须包含所有字段参数
        checkArgument(parameterMap.size() == fieldNames.length);
        
        int[][] indexMapping = new int[fieldNames.length][];
        
        for (int i = 0; i < fieldNames.length; i++) {
            
            String fieldName = fieldNames[i];
            
            //这行代码使用了 checkArgument 方法，用于检查在 SQL语句中是否存在名为 fieldName 的参数。
            // 如果不存在，则抛出一个异常，并在异常消息中包含参数名和 SQL 语句。
            checkArgument(parameterMap.containsKey(fieldName),
                    fieldName + " doesn't exist in the parameters of SQL statement: " + sql);
            
            // parameterMap.get(fieldName)返回一个元素类型为 List<Integer>的列表，
            // 表示名为 fieldName的参数在SQL语句中出现的所有位置。
            // 通过 stream()方法将列表转换为一个Stream对象，
            // 然后通过 mapToInt(v -> v) 将每个位置转换为整型值，并返回一个 IntStream对象。
            // 最后，通过 toArray()方法将 IntStream对象转换为一个整型数组，
            // 并将其存储在 indexMapping数组中。
            indexMapping[i] = parameterMap.get(fieldName).stream().mapToInt(v -> v).toArray();
        }

        // 初始化 FieldNamedPreparedStatementImpl类的实例。
        PreparedStatement preparedStatement = connection.prepareStatement(parsedSQL);
        System.out.println("测试结果：" + preparedStatement);
        //prepareStatement方法返回结果：
        // org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl@7689ddef
        System.out.println("----————————————————————————----");
        return new FieldNamedPreparedStatementImpl(connection.prepareStatement(parsedSQL), indexMapping);
    }

    /**
     * Parses a sql with named parameters. 使用命名参数解析 SQL。
     * 
     * The parameter-index mappings are put into the map, and the parsed sql is returned.
     * 将参数-索引映射放入 map中，并返回解析后的 sql。
     * 
     * parseNamedStatement是一个静态方法，用于解析带有命名参数的 SQL语句，
     *                    并将其转换为使用占位符的 SQL语句。
     * 
     * @param sql sql to parse
     * @param paramMap map to hold parameter-index mappings
     * @return the parsed sql
     */
    public static String parseNamedStatement(String sql, Map<String, List<Integer>> paramMap) {
        
        //首先，创建一个 StringBuilder对象 parsedSql，用于存储转换后的 SQL语句。
        StringBuilder parsedSql = new StringBuilder();
        
        int fieldIndex = 1; // SQL statement parameter index starts from 1
        int length = sql.length();
        
        //然后，遍历原始 SQL语句中的每个字符。
        for (int i = 0; i < length; i++) {
            
            //从字符串sql中获取索引为 i 的字符，并将其赋值给变量 c
            char c = sql.charAt(i);
            
            // 如果当前字符是冒号（:），则说明遇到了一个命名参数。
            if (':' == c) {
                
                // 接着，从当前位置开始向后查找，直到找到一个非 Java标识符字符为止。
                int j = i + 1;
                
                //使用 Character.isJavaIdentifierPart()方法来判断一个字符是否是 Java标识符字符。
                //定义变量 j 并初始化为 i + 1，表示从当前字符的下一个字符开始查找。
                //在循环中，首先检查 j是否小于字符串 sql的长度，并且字符 sql.charAt(j)是否是 Java标识符字符。
                while (j < length && Character.isJavaIdentifierPart(sql.charAt(j))) {
                    System.out.print("Java标识符字符"  + sql.charAt(j) + "\n");
                    
                    // 如果满足条件，则将变量 j 加 1，继续向后查找，
                    // 直到找到一个 非Java标识符字符为止。
                    j++;
                }
                //循环结束后，变量 j指向的位置是第一个非Java标识符字符的位置，
                // 也就是命名参数名称的末尾位置（不包括该字符）。
                
                //使用 substring(i + 1, j) 方法从字符串sql中截取命名参数名称，
                // 并将其赋值给变量 parameterName。
                String parameterName = sql.substring(i + 1, j);
                System.out.println("截取结果：" + parameterName);
                
                checkArgument(!parameterName.isEmpty(),
                        "Named parameters in SQL statement must not be empty.");
                
                // 将命名参数的名称和对应的位置添加到 paramMap中。
                
                // 首先，使用 computeIfAbsent()方法从 paramMap中获取命名参数 parameterName对应的值，
                // 如果该参数不存在，则会使用提供的 lambda表达式创建一个新的 ArrayList对象，并将其作为参数值。
                
                // computeIfAbsent方法用于根据指定的键执行计算，并将计算结果作为值存储在 Map中。
                // 如果 Map中已经存在该键的值，则不执行计算，直接返回已有的值。
                
                //接着，将 fieldIndex添加到获取到的值（即 ArrayList对象）中，
                //表示该命名参数在 SQL语句中的位置。
                paramMap.computeIfAbsent(parameterName, n -> new ArrayList<>()).add(fieldIndex);

                // 并将参数位置加 1。
                fieldIndex++;

                i = j - 1;

                // 同时在 parsedSql中添加一个占位符（?）
                parsedSql.append('?');
            } else {
                //如果当前字符不是冒号，则直接将其添加到parsedSql中。
                parsedSql.append(c);
            }
        }
        //最后，返回转换后的 SQL语句。
        System.out.print("转换后的SQL语句：" +  parsedSql.toString());
        return parsedSql.toString();
    }
}
